package com.maozzi.api.mq.handler;

import com.google.gson.Gson;
import com.maozzi.api.mq.model.BaseMsg;
import com.maozzi.api.mq.template.MsgTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;

/**
 * message 处理
 *
 * @author maozi
 */

@Slf4j
public abstract class MsgHandler<T extends BaseMsg> {
    /**
     * 默认重试次数
     */
    private final static int RETRY_TIMES = 3;

    private final static String PREFIX = "retry_";

    private final static Gson GSON = new Gson();

    @Resource
    private MsgTemplate msgTemplate;

    protected abstract void handleMsg(T msg) throws Exception;

    protected abstract void handleMaxRetry(T msg) throws Exception;

    /**
     * 是否需要根据业务规则过滤消息，去重逻辑可以在此处处理
     *
     * @param msg 待处理消息
     * @return true: 本次消息被过滤，false：不过滤
     */
    protected boolean filter(T msg) {
        return false;
    }

    protected abstract boolean isRetry();

    protected abstract boolean throwException();

    protected int getMaxRetry() {
        return RETRY_TIMES;
    }

    public void dispatchMsg(T msg) {
        log.info("消费者收到消息[{}]", GSON.toJson(msg));

        if (this.filter(msg)) {
            log.info("消息id{}不满足消费条件，已过滤。", msg.getKey());
            return;
        }
        try {
            // 超过最大重试次数时调用子类方法处理
            if (msg.getRetryTimes() > getMaxRetry()) {
                this.handleMaxRetry(msg);
                return;
            }

            long now = System.currentTimeMillis();
            this.handleMsg(msg);
            long costTime = System.currentTimeMillis() - now;
            log.info("消息{}消费成功，耗时[{}ms]", msg.getKey(), costTime);
        } catch (Exception e) {
            log.error("消息{}消费异常", msg.getKey(), e);

            // 是捕获异常还是抛出，由子类决定
            if (this.throwException()) {
                //抛出异常，由DefaultMessageListenerConcurrently类处理
                throw new RuntimeException(e);
            }

            //此时如果不开启重试机制，则默认ACK了
            if (this.isRetry()) {
                this.handleRetry(msg);
            }
        }
    }


    public void handleRetry(T msg) {
        RocketMQMessageListener annotation = this.getClass().getAnnotation(RocketMQMessageListener.class);
        if (annotation == null) {
            return;
        }

        String source = msg.getSource();
        if (!StringUtils.startsWithIgnoreCase(source, PREFIX)) {
            msg.setSource(PREFIX + source);
        }
        msg.setRetryTimes(msg.getRetryTimes() + 1);

        SendResult result = msgTemplate.send(annotation.topic(), annotation.selectorExpression(), msg);
        // 发送失败的处理就是不进行ACK，由RocketMQ重试
        if (result.getSendStatus() != SendStatus.SEND_OK) {
            throw new RuntimeException("重试消息发送失败");
        }
    }
}
